package com.example.rocket.config;

import com.example.rocket.annotation.NbMQMessageListener;
import com.example.rocket.consumer.AbstractNbMQConsumer;
import com.example.rocket.consumer.OnsNbMQConsumer;
import com.example.rocket.consumer.RocketNbMQConsumer;
import com.example.rocket.consumer.NbMqListener;
import com.example.rocket.core.ConsumeMode;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.scope.ScopedProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.util.StringUtils;

import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * @author TAO
 * @description: NbMq-consumer自动挂载
 * 主要是识别出被@NbMQMessageListener注解标注的类，然后注册为消费者监听器
 * @date 2021/12/21 4:14 下午
 */

@Slf4j
@Configuration
public class NbAutoConsumerConfiguration implements ApplicationContextAware, SmartInitializingSingleton {

    private ConfigurableApplicationContext applicationContext;

    private AtomicLong counter = new AtomicLong(0);

    private StandardEnvironment environment;

    private NbMqProperties properties;

    public NbAutoConsumerConfiguration(StandardEnvironment environment,NbMqProperties properties) {
        this.environment = environment;
        this.properties = properties;
    }


    /**
     * 实现ApplicationContextAware接口从写，详情请见：https://blog.csdn.net/CSDN877425287/article/details/115499665
     * @param applicationContext
     * @throws BeansException
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }


    /**
     * 查找所有被NbMQMessageListener标注的容器
     */
    @Override
    public void afterSingletonsInstantiated() {
        //得到所有标记的监听器
        Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(NbMQMessageListener.class).entrySet().stream().filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey()))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        //log.trace("beanName===>{}  bean===>{}", beanName, bean);

        //循环将监听器注册为Bean
        beans.forEach(this::registerContainer);
    }

    /**
     * 注册容器
     * @param beanName
     * @param bean
     */
    private void registerContainer(String beanName, Object bean) {
        log.trace("beanName===>{}  bean===>{}", beanName, bean);
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);

        //检验监听器是否有实现NbMqListener接口，确保存在onMessage方法，用于接受消费消息
        if (!NbMqListener.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(clazz + " 当前消费者监听器必须实现==> " + NbMqListener.class.getName());
        }

        //得到注解值
        NbMQMessageListener annotation = clazz.getAnnotation(NbMQMessageListener.class);

        //检验Consumer配置的合法性
        validate(annotation);

        //TODO 后期可以设置控制某个消费者是否开启

        //生成Bean名
        String containerBeanName = String.format("%s_%s", NbAutoProducerConfiguration.class.getName(),
                counter.incrementAndGet());

        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;

        //将Bean注册在容器中
        genericApplicationContext.registerBean(containerBeanName, AbstractNbMQConsumer.class,
                () -> createAbstractNbMQConsumer(containerBeanName, bean, annotation));

        //得到当前这个监听器
        AbstractNbMQConsumer container = genericApplicationContext.getBean(containerBeanName,
                AbstractNbMQConsumer.class);

        //判断当前监听器是否运行
        if (!container.isRunning()) {
            try {
                //启动监听器
                container.start();
            } catch (Exception e) {
                log.error("启动MQ消费者失败{}", container, e);
                throw new RuntimeException(e);
            }
        }
        log.info("将监听器注册到容器, 监听器Bean名字:{}, 容器Bean名称:{}", beanName, containerBeanName);
    }


    /**
     * 创建抽象消费者
     * @param name
     * @param bean
     * @param annotation
     * @return
     */
    private AbstractNbMQConsumer createAbstractNbMQConsumer(String name, Object bean, NbMQMessageListener annotation) {

        AbstractNbMQConsumer container = null;
        if ("rocket".equals(properties.getType())){
            container= new RocketNbMQConsumer();
        }else if ("ons".equals(properties.getType())){
            container= new OnsNbMQConsumer();
        }else{
            log.error("异常===>"+"既不是rocket也不是ons，你到底是啥玩意！！！");
        }

        //设置consumer属性值，这些值都是无序特殊处理的
        container.setOtherProperties(annotation);

        /**
         * environment.resolvePlaceholders ：https://blog.csdn.net/CSDN877425287/article/details/122102334
         * 从标准环境（environment）中获取注解NbMQMessageListener中的nameServer值，
         * nameServer在注解上如果没有配置，那么默认为${rmq.name-server:}，则根据${rmq.name-server:}在标准环境中匹配值
         * nameServer在注解上如果配置了，那么就是直接返回配置的值
         */
        String nameServer = environment.resolvePlaceholders(annotation.nameServer());
        nameServer = StringUtils.isEmpty(nameServer) ? properties.getNameServer() : nameServer;
        container.setNameServer(nameServer);

        String accessChannel = environment.resolvePlaceholders(annotation.accessChannel());
        if (!StringUtils.isEmpty(accessChannel)) {
            container.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }

        //将当前bean设置为监听器
        if (NbMqListener.class.isAssignableFrom(bean.getClass())) {
            container.setNbMqListener((NbMqListener) bean);
        }




        //TODO 消息类型转换完善
        //container.setMessageConverter(rocketMQMessageConverter.getMessageConverter());

        return container;
    }


    /**
     * Consumer合法性检验
     * @param annotation
     */
    private void validate(NbMQMessageListener annotation) {
        if (annotation.consumeMode() == ConsumeMode.ORDERLY &&
                annotation.messageModel() == MessageModel.BROADCASTING) {
            throw new BeanDefinitionValidationException(
                    "Bad annotation definition in @RocketMQMessageListener, messageModel BROADCASTING does not support ORDERLY message!");
        }
    }

}
